package com.lunsheng.train

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * @Description
  * @author zhonglunsheng
  * @create 2020-02-10 23:20
  */
object StateWordCount {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[3]").setAppName("WorldCountV2")
    val ssc = new StreamingContext(conf, Seconds(5))
    ssc.checkpoint("E:/temp/checkpoint")
    val lines = ssc.textFileStream("D:/temp")

    val result = lines.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_)

    val state = result.updateStateByKey[Int](updateFunction _)
    state.print()


    ssc.start()
    ssc.awaitTermination()
  }

  def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
    val currentValue = newValues.sum
    val preValue = runningCount.getOrElse(0)
    val newCount = currentValue + preValue
    Some(newCount)
  }
}
